package com.huatai.datacenter.job;

import com.huatai.datacenter.service.AllURLService;
import com.huatai.datacenter.service.KafkaConsumerService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;

import java.time.Duration;

/**
 * @author Lion
 * @date 2023/6/13  9:23
 */
@Slf4j
public class KafkaListenerJob implements Runnable{

	@Autowired
	private KafkaConsumerListener kafkaConsumerListener;

	@Autowired
	private KafkaConsumerService kafkaConsumerService;

	private KafkaConsumer<String, String> kafkaConsumer;

	private long modelId;

	private String tableName;

	private String structure;

	private String qualityRule;
	private String modelName;

	/**
	 * @author Lion
	 * @description 构造函数
	 * @param kafkaConsumerListener
	 * @param kafkaConsumer
	 * @param modelId
	 * @param tableName
	 * @param structure
	 * @param qualityRule
	 * @return
	 */
	public KafkaListenerJob(KafkaConsumerListener kafkaConsumerListener, KafkaConsumer<String, String> kafkaConsumer, long modelId, String modelName, String tableName, String structure, String qualityRule) {

		this.kafkaConsumerListener = kafkaConsumerListener;
		this.kafkaConsumer = kafkaConsumer;
		this.modelId = modelId;
		this.tableName = tableName;
		this.structure = structure;
		this.qualityRule = qualityRule;
		this.modelName = modelName;
		this.kafkaConsumerService = kafkaConsumerService;
	}

	@Override
	public void run() {

		while (true){
			try {
				ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
				for (ConsumerRecord<String, String> record : records) {
					//保存实时数据到数据库中 MySQL的
					kafkaConsumerListener.listen(record, modelId, modelName, tableName, structure, qualityRule);

					//保存实时数据到数据库中 MongoDB的
//					kafkaConsumerListener.listenMongoDB(record, modelId, modelName, tableName, structure, qualityRule);
				}
			} catch (Exception e) {
				//关闭订阅
				kafkaConsumer.close();
				//log.error(e.getMessage());
				continue;
			}
		}
	}
}
